package com.spzx.common.rabbit.config;

import com.alibaba.fastjson2.JSON;
import com.spzx.common.rabbit.entity.GuiguCorrelationData;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

/**
 * 应用启动初始化监听器(应用程序启动好就执行初始化回调)
 */
@Slf4j
@Component
public class RabbitInitConfigApplicationListener implements ApplicationListener<ApplicationReadyEvent> {

    @Autowired
    RabbitTemplate rabbitTemplate;

    @Autowired
    RedisTemplate redisTemplate;

    @Override
    public void onApplicationEvent(ApplicationReadyEvent event) {
        this.setupCallbacks();
    }

    /**
     * 初始化：设置rabbitTemplate确认回调函数和回退回调函数。
     */
    private void setupCallbacks() {

        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            /**
             * 确认模式：发送消息不管成功与否，都会回调！
             * @param correlationData 关联数据.
             * @param ack 是否确认    true: 表示交换机收到消息了。false：表示交换机没收到消息。
             * @param cause 消息发送失败的原因。
             */
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                log.info("correlationData=" + correlationData); //发送消息时如果携带关联数据，回调时会返回这个关联数据。否则就是空。
                if (ack) { //表示交换机收到消息了
                    log.info("表示交换机[收到]消息了");
                } else {
                    log.info("表示交换机[没收到]消息了");
                    log.info("cause=" + cause);

                    //重发消息
                    retrySendMsg(correlationData);
                }
            }
        });


        rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
            /**
             * 回退模式：消息由交换机成功转发到队列不进行回退；交换机转发队列失败则回退。
             * @param returned 回退消息和元信息(头信息等)
             */
            @Override
            public void returnedMessage(ReturnedMessage returned) {
                log.info("returned=" + returned);
                //从缓存中获取关联数据，进行消息重发
                /*
                returned=ReturnedMessage [message=(Body:'hello, confirm' MessageProperties
                [headers={spring_returned_message_correlation=f2c00dba89584c95bb564de94b661588},
                contentType=text/plain, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT,
                priority=0, deliveryTag=0]), replyCode=312, replyText=NO_ROUTE, exchange=spzx.test, routingKey=xxx]
                 */
                String correlationId = returned.getMessage()
                        .getMessageProperties().getHeader("spring_returned_message_correlation");
                String correlationDataStr = (String) redisTemplate.opsForValue().get(correlationId);
                GuiguCorrelationData correlationData = JSON.parseObject(correlationDataStr, GuiguCorrelationData.class);

                //方案一： 是延迟消息就不去重试；不是延迟消息才会重试
                /*if(correlationData.isDelay()){
                    return;
                }*/

                //重发消息
                retrySendMsg(correlationData);
            }
        });

    }


    /**
     * 消息重新发送
     *
     * @param correlationData
     */
    private void retrySendMsg(CorrelationData correlationData) {

        GuiguCorrelationData guiguCorrelationData = (GuiguCorrelationData) correlationData;

        //重试3次； 不包括首次
        int retryCount = guiguCorrelationData.getRetryCount();

        if (retryCount >= 3) {
            //超过最大重试次数
            log.error("生产者超过最大重试次数，将失败的消息存入数据库用由人工处理；给管理员发送邮件；给管理员发送短信；");
            return;
        }

        log.info("消息重试："+retryCount);

        guiguCorrelationData.setRetryCount(++retryCount);

        //方案二: 延迟消息和非延迟消息发送的方式不同。单独处理。
        if(guiguCorrelationData.isDelay()){
            //发送消息同时，把关联数据也发送。回退时需要把关联数据也进行回退，根据关联数据就可以进行消息重复。
            rabbitTemplate.convertAndSend(
                    guiguCorrelationData.getExchange(),
                    guiguCorrelationData.getRoutingKey(),
                    guiguCorrelationData.getMessage(),
                    messageObj -> {
                        messageObj.getMessageProperties().setDelay(guiguCorrelationData.getDelayTime() * 1000); //单位毫秒
                        return messageObj;
                    },
                    correlationData);
        }else{
            //重发
            rabbitTemplate.convertAndSend(guiguCorrelationData.getExchange(),
                    guiguCorrelationData.getRoutingKey(),
                    guiguCorrelationData.getMessage(),
                    guiguCorrelationData);
        }

        //覆盖缓存数据。
        redisTemplate.opsForValue().set(correlationData.getId(),
                JSON.toJSONString(correlationData),5, TimeUnit.MINUTES);
    }
}
